package dongze.com.ckdj.service.uploader;

import android.util.Log;

import dongze.com.ckdj.persistence.dao.State;
import dongze.com.ckdj.service.ImageUploadTask;
import dongze.com.ckdj.util.MyLog;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.disposables.Disposable;


/**
 * Created by fanqwz on 2017/8/16.
 */

/**
 *@param <D> 依赖的{@link Uploader}的返回值,即此{@link Uploader}持有的{@link Uploader.Worker}对象接受的参数
 * @param <T> 此{@link Uploader}的返回值*/
public abstract class Uploader<T, D> {
    private static final String TAG = Uploader.class.getSimpleName();
    final Scheduler subscribeThread;
    final Scheduler observerThread;
    private Observable observable;
    private Observer<T> observer;
    private Uploader dependsOn;
    boolean isStarted = false;

    abstract Worker<T, D> getWorker();

    public Uploader(Scheduler subscribeThread, Scheduler observerThread) {
        this.subscribeThread = subscribeThread;
        this.observerThread = observerThread;

        observable = Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(ObservableEmitter<T> e) throws Exception {
                MyLog.d(TAG, "subscribe: " + e.getClass().getSimpleName());
                getWorker().doWork(e);
            }
        });
    }

    public <E> void dependsOn(Uploader<D, E> uploader) {
        dependsOn = uploader;
        uploader.observe(new Observer<D>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(D value) {
                getWorker().accept(value);
            }

            @Override
            public void onError(Throwable e) {
                MyLog.e(Uploader.this.getClass().getSimpleName(), "dependsOn::onError: ", e);

                if (observer != null)
                    observer.onError(e);
            }

            @Override
            public void onComplete() {
                startThis();
            }
        });
    }

    public void observe(Observer observer) {
        this.observer = observer;
    }

    private void startThis() {
        if (isStarted)
            return;

        isStarted = true;

        if (observer != null) {
            observable.subscribeOn(subscribeThread)
                    .observeOn(observerThread)
                    .subscribeWith(observer);
        } else {
            MyLog.e(getClass().getSimpleName(), "start: observer null");
            observable.subscribeOn(subscribeThread)
                    .observeOn(observerThread)
                    .subscribe();
        }
    }

    public void start() {
        if (dependsOn != null)
            dependsOn.start();
        else
            startThis();
    }

    public void cancel() {
        if (dependsOn != null)
            dependsOn.cancel();

        MyLog.d(TAG, "cancel: " + this);
        observable.unsubscribeOn(subscribeThread);
    }

    public static abstract class Worker<T, D> {
        abstract public void accept(D depends);

        abstract public void doWork(ObservableEmitter<T> e);
    }
}
